/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package dsmsforkjoin;

import tuplegenerator.MeterUser;
import tuplegenerator.SingleTuple;
import dsmsforkjoin.joins.*;


public class RunMe {
    
    public static int MAXIMUM_FORKS = 5;
    public static int TOTAL_TUPLES = 10000;
    
    public static void main(String[] args) {
        RunMe rm = new RunMe();
        rm.run();
    }
    
    private int finished = 2;
    private FJoin<SingleTuple> fj;
    private FJoin<MeterUser> fj2;
    
    public RunMe() {
        
    }
    
    public void run() {
        this.finished = 2;
        
        ForkList<IForkQueue<SingleTuple>, SingleTuple> stqueue = new ForkListSimpleQueue<SingleTuple>(MAXIMUM_FORKS);
        ForkList<IForkQueue<MeterUser>, MeterUser> mtqueue = new ForkListSimpleQueue<MeterUser>(MAXIMUM_FORKS);
        
        fj = new FJoin<SingleTuple>(stqueue);
        fj2 = new FJoin<MeterUser>(mtqueue);
        
        InputHandlerMeterReadings<SingleTuple> mreads = new InputHandlerMeterReadings<SingleTuple>(this, fj);
        InputHandlerUserMeters<MeterUser> users = new InputHandlerUserMeters<MeterUser>(this, fj2);
        
        Thread t = new Thread(mreads);
        Thread tt = new Thread(users);
        t.start();
        tt.start();
    }
    
    public synchronized void reduce() {
        this.finished--;
        
        if (this.finished == 0) {
            System.out.println("===================\nIT'S ALL OVER\n=============================");
            
            long op1 = System.currentTimeMillis();
            IJoinOperator<SingleTuple, MeterUser> jop = new SortMergeJoinOperator<SingleTuple, MeterUser>();
            jop.doJoin(fj, fj2);
            long op2 = System.currentTimeMillis();
            
            System.out.println("Time taken to complete join " + jop.getClass().getName() + " is:");
            System.out.println((op2-op1) + "ms");
        }
    }
    
}
